/**********************************************************
*****************订阅端程序subscriber.cpp*******************
***********************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <cstring>
#include <chrono>
#include <atomic>
#include <mutex>
#include <iomanip>
#include <algorithm>

/* IDL_TypeSupport.h中包含所有依赖的头文件 */
#include "IDL_TypeSupport.h"
#include "md5.h"             // 自定义MD5实现

using namespace std;
using namespace std::chrono;

// 全局统计变量
atomic<unsigned long long> total_recv(0);       // 总接收字节
atomic<unsigned long long> total_lost(0);      // 总丢包数
atomic<unsigned long long> last_seq_num(0);    // 上一个序列号
mutex print_mutex;                            // 输出锁

template<typename T>
const T& clamp(const T& value, const T& low, const T& high) {
    return (value < low) ? low : (value > high) ? high : value;
}

/* 删除所有实体 */
static int subscriber_shutdown(DomainParticipant* participant) {
    ReturnCode_t retcode;
    int status = 0;

    if (participant != nullptr) {
        retcode = participant->delete_contained_entities();
        if (retcode != RETCODE_OK) {
            fprintf(stderr, "删除实体失败: %d\n", retcode);
            status = -1;
        }

        retcode = DomainParticipantFactory::get_instance()->delete_participant(participant);
        if (retcode != RETCODE_OK) {
            fprintf(stderr, "删除参与者失败: %d\n", retcode);
            status = -1;
        }
    }
    return status;
}

/* 数据接收监听器 */
class TestDataListener : public DataReaderListener {
public:
    explicit TestDataListener(DataWriter* writer = nullptr) : writer_(writer) {}

    void on_data_available(DataReader* reader) override {
        TestDataDataReader* data_reader = TestDataDataReader::narrow(reader);
        TestDataSeq data_seq;
        SampleInfoSeq info_seq;

        // 读取数据
        ReturnCode_t retcode = data_reader->take(
            data_seq, info_seq,
            LENGTH_UNLIMITED,
            ANY_SAMPLE_STATE,
            ANY_VIEW_STATE,
            ANY_INSTANCE_STATE
        );

        if (retcode != RETCODE_OK) {
            cerr << "读取数据失败: " << retcode << endl;
            return;
        }

        // 处理每个数据样本
        for (int i = 0; i < data_seq.length(); ++i) {
            if (!info_seq[i].valid_data) continue;

            const TestData& data = data_seq[i];

            // === 1. 计算MD5校验值 ===
            MD5 md5;
            sequence<octet> non_const_payload = data.payload;  
            md5.update(non_const_payload.get_buffer(), non_const_payload.length());
            string recv_md5 = md5.hexdigest();

            // 校验MD5是否一致
            if (strncmp(recv_md5.c_str(), data.md5, 32) != 0) {
                lock_guard<mutex> lock(print_mutex);
                cerr << "[ERROR] MD5校验失败! Seq: " << data.seq_num << endl;
            }

            // === 2. 统计吞吐量和丢包 ===
            total_recv += data.payload.length();

            // 计算丢包数（基于序列号连续性）
            if (data.seq_num > last_seq_num + 1 && last_seq_num != 0) {
                total_lost += (data.seq_num - last_seq_num - 1);
            }
            last_seq_num = data.seq_num;

            // === 3. 时延测试：发送响应 ===
            if (writer_ != nullptr) {
                TestData response;
                response.seq_num = data.seq_num;
                response.send_time = data.send_time;
                response.payload.length(1);  // 响应数据最小化

                TestDataDataWriter::narrow(writer_)->write(response, HANDLE_NIL);
            }
        }

        // 归还数据
        data_reader->return_loan(data_seq, info_seq);
    }

private:
    DataWriter* writer_;  // 用于时延测试的响应发送
};

/* 主测试函数 */
extern "C" int subscriber_main(int domainId, int test_mode, int data_size, int loop_count) {
    DomainParticipant* participant = nullptr;
    Subscriber* subscriber = nullptr;
    Publisher* publisher = nullptr;
    Topic* topic = nullptr;
    DataReader* data_reader = nullptr;
    DataWriter* response_writer = nullptr;
    TestDataListener* listener = nullptr;

    /* 1. 创建域参与者 */
    participant = DomainParticipantFactory::get_instance()->create_participant(
        domainId, PARTICIPANT_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    if (!participant) {
        cerr << "创建participant失败" << endl;
        return -1;
    }

    /* 2. 创建订阅者（和发布者用于时延测试） */
    subscriber = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    if (!subscriber || !publisher) {
        cerr << "创建subscriber/publisher失败" << endl;
        return subscriber_shutdown(participant);
    }

    /* 3. 注册数据类型 */
    const char* type_name = TestDataTypeSupport::get_type_name();
    if (TestDataTypeSupport::register_type(participant, type_name) != RETCODE_OK) {
        cerr << "注册类型失败" << endl;
        return subscriber_shutdown(participant);
    }

    /* 4. 创建主题 */
    topic = participant->create_topic(
        "Example TestData",
        type_name,
        TOPIC_QOS_DEFAULT,
        nullptr,
        STATUS_MASK_NONE
    );
    if (!topic) {
        cerr << "创建topic失败" << endl;
        return subscriber_shutdown(participant);
    }

    /* 5. 创建DataReader */
    listener = new TestDataListener();
    data_reader = subscriber->create_datareader(
        topic,
        DATAREADER_QOS_DEFAULT,
        listener,
        STATUS_MASK_ALL
    );
    if (!data_reader) {
        cerr << "创建DataReader失败" << endl;
        return subscriber_shutdown(participant);
    }

    /* 6. 时延测试需要创建响应DataWriter */
    if (test_mode == 3) {
        response_writer = publisher->create_datawriter(
            topic,
            DATAWRITER_QOS_DEFAULT,
            nullptr,
            STATUS_MASK_NONE
        );
        if (!response_writer) {
            cerr << "创建响应DataWriter失败" << endl;
            return subscriber_shutdown(participant);
        }
        listener = new TestDataListener(response_writer); // 重置监听器
    }

    /* 7. 主循环（每秒显示统计信息） */
    auto last_print = high_resolution_clock::now();
    while (true) {
        this_thread::sleep_for(1s);

        // 吞吐量统计
        if (test_mode == 2 || test_mode == 3) {
            auto now = high_resolution_clock::now();
            if (duration_cast<seconds>(now - last_print).count() >= 1) {
                lock_guard<mutex> lock(print_mutex);

                // 计算吞吐量和丢包率
                double mbps = (total_recv.load() * 8.0) / 1e6;
                unsigned long long total_packets = last_seq_num.load();
                double loss_rate = (total_lost.load() * 100.0) / (total_packets + total_lost.load());

                cout << "[RX] 吞吐量: " << fixed << setprecision(2) << mbps << " Mbps | "
                    << "丢包率: " << setprecision(2) << loss_rate << "%" << endl;

                // 重置统计
                total_recv = 0;
                total_lost = 0;
                last_print = now;
            }
        }
    }

    /* 8. 资源清理（实际不会执行到此处） */
    delete listener;
    return subscriber_shutdown(participant);
}

/* 程序入口 */
int main(int argc, char* argv[]) 
{
    if (argc < 5) {
        cerr << "用法: " << argv[0] << " <域ID> <测试模式> <数据大小> <循环次数>\n"
            << "  测试模式: 1-功能 2-吞吐量 3-时延\n"
            << "  数据大小: 1-1048576 字节（需与发布端一致）\n"
            << "  循环次数: 发布端迭代次数（时延测试需匹配）\n";
        return -1;
    }

    // 参数解析（仅用于显示，实际由发布端控制）
    const int domain_id = atoi(argv[1]);
    const int test_mode = clamp(atoi(argv[2]), 1, 3);
    const int data_size = clamp(atoi(argv[3]), 1, 1048576);
    const int loop_count = max(atoi(argv[4]), 1);

    return subscriber_main(domain_id, test_mode, data_size, loop_count);
}